package com.autoai.tripflink

import java.util.Properties

import com.autoai.model.StandCarData
import net.sf.json.JSONObject
import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.java.aggregation.Aggregations
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011

object flinkWindowTrips {
  def main(args: Array[String]): Unit = {

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "10.30.23.45:9092")
    properties.setProperty("group.id", "guroupuaidi111")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    val stream = env.addSource(new FlinkKafkaConsumer011[String]("gwSceneReadDataTest", new SimpleStringSchema(), properties).setStartFromGroupOffsets()).uid("yeah")

    val standData = stream.map(
      (data: String) => {

        val result: JSONObject = JSONObject.fromObject(data)
        val vin = result.get("vin").toString
        val userid = result.get("userId").toString
        val repoerTime = result.get("reportTime").toString.toLong
        val VehTotDistance = result.get("VehTotDistance").toString.toDouble
        val lon = result.get("lon").toString.toDouble
        val lat = result.get("lat").toString.toDouble
        val SystemPowerMod = result.get("SystemPowerMod").toString.toInt

        StandCarData(vin, userid, repoerTime, VehTotDistance, lon, lat, SystemPowerMod)


      }
    ).keyBy(d => d.vin)
      .window(EventTimeSessionWindows.withGap(Time.milliseconds(500)))




  }

}
class MyReducer extends ReduceFunction[StandCarData]{
  override def reduce(value1: StandCarData, value2: StandCarData): StandCarData = {
    if(value1.mileage < value2.mileage) {
      
    }
  }
}
